package org.finesys.common.netty.websocket.stomp.handler;

import static io.netty.handler.codec.http.HttpHeaderNames.AUTHORIZATION;
import static io.netty.handler.codec.stomp.StompHeaders.ACCEPT_VERSION;
import static io.netty.handler.codec.stomp.StompHeaders.DESTINATION;
import static io.netty.handler.codec.stomp.StompHeaders.HEART_BEAT;
import static io.netty.handler.codec.stomp.StompHeaders.ID;
import static io.netty.handler.codec.stomp.StompHeaders.LOGIN;
import static io.netty.handler.codec.stomp.StompHeaders.MESSAGE;
import static io.netty.handler.codec.stomp.StompHeaders.MESSAGE_ID;
import static io.netty.handler.codec.stomp.StompHeaders.RECEIPT;
import static io.netty.handler.codec.stomp.StompHeaders.RECEIPT_ID;
import static io.netty.handler.codec.stomp.StompHeaders.SERVER;
import static io.netty.handler.codec.stomp.StompHeaders.SUBSCRIPTION;
import static io.netty.handler.codec.stomp.StompHeaders.VERSION;
import static org.finesys.common.netty.websocket.constants.WebSocketAttributeKeyConstants.DESTINATIONS;
import static org.finesys.common.netty.websocket.constants.WebSocketAttributeKeyConstants.SESSION_KEY;
import static org.finesys.common.netty.websocket.constants.WebSocketAttributeKeyConstants.USER_ID;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.finesys.common.netty.websocket.stomp.constants.StompVersion;
import org.finesys.common.netty.websocket.stomp.entity.StompSubscription;
import org.finesys.common.netty.websocket.stomp.support.StompWebSocketEndpointEventServer;
import org.finesys.common.netty.websocket.support.NettyChannelMap;
import org.finesys.common.netty.websocket.support.NettySession;
import org.finesys.common.netty.websocket.support.WebSocketEndpointConfig;
import org.finesys.common.netty.websocket.support.WsPathMatcher;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.DecoderResult;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.stomp.DefaultStompFrame;
import io.netty.handler.codec.stomp.StompCommand;
import io.netty.handler.codec.stomp.StompFrame;
import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.EventExecutorGroup;

/**
 * 自定义处理协议内容
 * <p>
 * <p>
 * 参考：
 * https://github.com/CG-Lin/whales-netty
 * https://gitee.com/shiquxingyang/websocket-server
 */
public class StompServerHandler extends SimpleChannelInboundHandler<StompFrame> {

    // WebSocket事件服务器
    private final StompWebSocketEndpointEventServer webSocketEndpointEventServer;

    // WebSocket端点配置
    private final WebSocketEndpointConfig webSocketEndpointConfig;

    // 事件执行器组
    private final EventExecutorGroup eventExecutorGroup;


    // 存储聊天目的地的订阅信息
    private final ConcurrentMap<String, Set<StompSubscription>> chatDestinations = new ConcurrentHashMap<>();

    /**
     * 构造方法，初始化WebSocket事件服务器、端点配置和事件执行器组。
     *
     * @param webSocketEndpointEventServer WebSocket事件服务器实例。
     * @param webSocketEndpointConfig      WebSocket端点配置对象。
     * @param eventExecutorGroup           事件执行器组，用于处理异步任务。
     * @param isCors                       是否启用跨源资源共享（CORS）。
     */
    public StompServerHandler(StompWebSocketEndpointEventServer webSocketEndpointEventServer, WebSocketEndpointConfig webSocketEndpointConfig, EventExecutorGroup eventExecutorGroup) {
        this.webSocketEndpointEventServer = webSocketEndpointEventServer;
        this.webSocketEndpointConfig = webSocketEndpointConfig;
        this.eventExecutorGroup = eventExecutorGroup;

    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {
            //获取stomp协议版本
            StompVersion stompVersion = StompVersion.findBySubProtocol(((WebSocketServerProtocolHandler.HandshakeComplete) evt).selectedSubprotocol());
            //设置stomp协议版本
            ctx.channel().attr(StompVersion.CHANNEL_ATTRIBUTE_KEY).setIfAbsent(stompVersion);
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    /**
     * 处理从客户端接收到的STOMP帧。
     *
     * @param ctx          通道处理器上下文，包含通道、管道等信息。
     * @param inboundFrame 接收到的STOMP帧。
     * @throws Exception 解码或处理帧时可能抛出的异常。
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, StompFrame inboundFrame) throws Exception {
        DecoderResult decoderResult = inboundFrame.decoderResult();
        if (decoderResult.isFailure()) {
            sendErrorFrame("rejected frame", decoderResult.toString(), ctx);
            return;
        }

        switch (inboundFrame.command()) {
            case STOMP:
            case CONNECT:
                onConnect(ctx, inboundFrame);
                break;
            case DISCONNECT:
                onDisconnect(ctx, inboundFrame);
                break;
            case SUBSCRIBE:
                onSubscribe(ctx, inboundFrame);
                break;
            case UNSUBSCRIBE:
                onUnsubscribe(ctx, inboundFrame);
                break;
            case SEND:
                onSend(ctx, inboundFrame);
                break;
            case ACK:
                onAck(ctx, inboundFrame);
                break;
            default:
                sendErrorFrame("unsupported command",
                        "Received unsupported command " + inboundFrame.command(), ctx);
        }
    }

    /**
     * 接收到一个CONNECT帧需要回复一个CONNECTED帧
     *
     * @param ctx
     * @param inboundFrame
     */
    private void onConnect(ChannelHandlerContext ctx, StompFrame inboundFrame) {
        String acceptVersions = inboundFrame.headers().getAsString(ACCEPT_VERSION);
        String loginUserId = inboundFrame.headers().getAsString(LOGIN);
        //版本信息
        StompVersion handshakeAcceptVersion = ctx.channel().attr(StompVersion.CHANNEL_ATTRIBUTE_KEY).get();
        if (acceptVersions == null || !acceptVersions.contains(handshakeAcceptVersion.getVersion())) {
            sendErrorFrame("invalid version",
                    "Received invalid version, expected " + handshakeAcceptVersion.getVersion(), ctx);
            return;
        }
        //用户认证信息
        //@todo 用户校验
        String authorization = inboundFrame.headers().getAsString(AUTHORIZATION);
        if (authorization == null) {
            sendErrorFrame("invalid credentials",
                    "Received invalid credentials, expected authorization header", ctx);
            return;
        }
        // 获取路径匹配模式
        String pattern = inboundFrame.headers().getAsString(DESTINATION);
        if (pattern == null) {
            sendErrorFrame("not found destination",
                    "Received invalid destination, expected authorization header", ctx);
            return;
        }

        // 处理握手前的回调
        //String subprotocols =
        handleBeforeHandshake(ctx, inboundFrame, pattern);
        //加入队列
        NettySession nettySession = new NettySession(ctx.channel());
        NettyChannelMap.getInstance().put(loginUserId, nettySession);
        ctx.channel().attr(SESSION_KEY).set(nettySession);
        //给连接设置userId
        ctx.channel().attr(USER_ID).set(loginUserId);
        //连接关闭时的兜底操作
        ctx.channel().closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                NettyChannelMap.getInstance().remove(loginUserId);
            }
        });
        //回复一个CONNECTED帧
        StompFrame connectedFrame = new DefaultStompFrame(StompCommand.CONNECTED);
        connectedFrame.headers()
                .set(VERSION, handshakeAcceptVersion.getVersion())
                .set(SERVER, "Netty-Server")
                .set(HEART_BEAT, "0,0");
        ctx.writeAndFlush(connectedFrame);
    }

    /**
     * 处理断开连接请求
     *
     * @param ctx          ChannelHandlerContext对象，表示当前连接的上下文信息
     * @param inboundFrame 收到的STOMP帧对象
     */
    private static void onDisconnect(ChannelHandlerContext ctx, StompFrame inboundFrame) {
        String receiptId = inboundFrame.headers().getAsString(RECEIPT);
        if (receiptId == null) {
            ctx.close();
            return;
        }
        StompFrame receiptFrame = new DefaultStompFrame(StompCommand.RECEIPT);
        receiptFrame.headers().set(RECEIPT_ID, receiptId);
        ctx.writeAndFlush(receiptFrame).addListener(ChannelFutureListener.CLOSE);
    }


    /**
     * 处理订阅请求
     *
     * @param ctx          ChannelHandlerContext对象，表示当前连接的上下文信息
     * @param inboundFrame 收到的STOMP帧对象
     */
    private void onSubscribe(ChannelHandlerContext ctx, StompFrame inboundFrame) {
        String destination = inboundFrame.headers().getAsString(DESTINATION);
        String subscriptionId = inboundFrame.headers().getAsString(ID);
        //String loginUserId = inboundFrame.headers().getAsString(LOGIN);

        if (destination == null || subscriptionId == null) {
            sendErrorFrame("missed header", "Required 'destination' or 'id' header missed", ctx);
            return;
        }
        Set<StompSubscription> subscriptions = chatDestinations.get(destination);
        if (subscriptions == null) {
            subscriptions = new HashSet<StompSubscription>();
            Set<StompSubscription> previousSubscriptions = chatDestinations.putIfAbsent(destination, subscriptions);
            if (previousSubscriptions != null) {
                subscriptions = previousSubscriptions;
            }
        }

        final StompSubscription subscription = new StompSubscription(subscriptionId, destination, ctx.channel());
        if (subscriptions.contains(subscription)) {
            sendErrorFrame("duplicate subscription",
                    "Received duplicate subscription id=" + subscriptionId, ctx);
            return;
        }

        subscriptions.add(subscription);
        //设置订阅信息到channel的属性中
        Set<StompSubscription> destList = Optional.ofNullable(ctx.channel().attr(DESTINATIONS).get())
                .orElse(new HashSet<>());
        destList.add(subscription);
        ctx.channel().attr(DESTINATIONS).set(destList);
        //断开连接时移除订阅信息
        ctx.channel().closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                chatDestinations.get(subscription.getDestination()).remove(subscription);
            }
        });
        //发送回执帧
        String receiptId = inboundFrame.headers().getAsString(RECEIPT);
        if (receiptId != null) {
            StompFrame receiptFrame = new DefaultStompFrame(StompCommand.RECEIPT);
            receiptFrame.headers().set(RECEIPT_ID, receiptId);
            ctx.writeAndFlush(receiptFrame);
        }
    }

    /**
     * 取消订阅
     *
     * @param ctx          ChannelHandlerContext对象，表示当前连接的上下文信息
     * @param inboundFrame 接收到的STOMP帧对象
     */
    private void onUnsubscribe(ChannelHandlerContext ctx, StompFrame inboundFrame) {
        String subscriptionId = inboundFrame.headers().getAsString(SUBSCRIPTION);
        for (Map.Entry<String, Set<StompSubscription>> entry : chatDestinations.entrySet()) {
            Iterator<StompSubscription> iterator = entry.getValue().iterator();
            while (iterator.hasNext()) {
                StompSubscription subscription = iterator.next();
                if (subscription.getId().equals(subscriptionId) && subscription.getChannel().equals(ctx.channel())) {
                    iterator.remove();
                    return;
                }
            }
        }
    }

    /**
     * 客户端发送消息到服务端
     *
     * @param ctx
     * @param inboundFrame
     */
    private void onSend(ChannelHandlerContext ctx, StompFrame inboundFrame) {
        String destination = inboundFrame.headers().getAsString(DESTINATION);
        if (destination == null) {
            sendErrorFrame("missed header", "required 'destination' header missed", ctx);
            return;
        }
        System.out.println("发送消息" + inboundFrame.content().toString());
    }

    /**
     * 确认这个消息已经被客户端消费了
     *
     * @param ctx
     * @param inboundFrame
     */
    private void onAck(ChannelHandlerContext ctx, StompFrame inboundFrame) {
        String msgId = inboundFrame.headers().getAsString(MESSAGE_ID);
        String subscriptionId = inboundFrame.headers().getAsString(SUBSCRIPTION);
        if (msgId == null || subscriptionId == null) {
            sendErrorFrame("missed header", "Required  'id' || 'subscriptionId' header missed", ctx);
            return;
        }
        StompFrame ackFrame = new DefaultStompFrame(StompCommand.ACK);
        ackFrame.headers().set(MESSAGE_ID, msgId);
        for (Map.Entry<String, Set<StompSubscription>> entry : chatDestinations.entrySet()) {
            for (StompSubscription subscription : entry.getValue()) {
                if (subscription.getId().equals(subscriptionId) && subscription.getChannel().equals(ctx.channel())) {
                    ctx.writeAndFlush(ackFrame);
                }
            }
        }
//        pushService.delMsgForSupplement(ctx.channel().attr(USER_ID).get(), dest, ackFrame);
    }

    /**
     * 发送完error帧后要关闭连接
     *
     * @param message
     * @param description
     * @param ctx
     */
    private static void sendErrorFrame(String message, String description, ChannelHandlerContext ctx) {
        StompFrame errorFrame = new DefaultStompFrame(StompCommand.ERROR);
        if (description != null) {
            errorFrame.content().writeCharSequence(description, CharsetUtil.UTF_8);
        }
        errorFrame.headers().set(MESSAGE, message);
        ctx.writeAndFlush(errorFrame).addListener(ChannelFutureListener.CLOSE);
    }

    /**
     * 获取与WebSocket路径匹配器对应的路径模式
     *
     * @param context            通道处理器上下文
     * @param queryStringDecoder 查询字符串解码器
     * @return 匹配到的路径模式，如果没有匹配到则返回null
     */
    private String getPathMatcherPattern(ChannelHandlerContext context, QueryStringDecoder queryStringDecoder) {
        // 获取通道处理器上下文中的通道
        Channel channel = context.channel();

        // 获取WebSocket端点事件服务器中的所有路径匹配器
        Set<WsPathMatcher> pathMatcherSet = webSocketEndpointEventServer.getPathMatchers();

        // 遍历所有路径匹配器
        for (WsPathMatcher wsPathMatcher : pathMatcherSet) {
            // 判断路径匹配器是否匹配并提取查询字符串解码器和通道
            if (wsPathMatcher.matchAndExtract(queryStringDecoder, channel)) {
                // 返回匹配到的路径模式
                return wsPathMatcher.getPattern();
            }
        }

        // 如果没有匹配到任何路径模式，则返回null
        return null;
    }

    /**
     * 在WebSocket握手之前进行处理
     *
     * @param context    通道处理器上下文
     * @param stompFrame 完整的HTTP请求
     * @param pattern    WebSocket路径模式
     * @return WebSocket子协议，如果没有则为null
     */
    private String handleBeforeHandshake(ChannelHandlerContext context, StompFrame stompFrame, String pattern) {
        String subprotocols = null;

        // 检查是否有WebSocket握手前的处理事件
        if (webSocketEndpointEventServer.hasBeforeHandshake(context.channel(), pattern)) {
            // 执行WebSocket握手前的处理事件
            webSocketEndpointEventServer.doBeforeHandshake(context.channel(), stompFrame, pattern);

            // 检查通道是否仍然活跃
            if (!context.channel().isActive()) {
                return null;
            }

            // 定义WebSocket子协议属性键
            AttributeKey<String> subprotocolsKey = AttributeKey.valueOf("subprotocols");

            // 检查通道是否有子协议属性
            if (context.channel().hasAttr(subprotocolsKey)) {
                // 获取子协议值
                subprotocols = context.channel().attr(subprotocolsKey).get();
            }
        }

        // 返回子协议值，如果没有则为null
        return subprotocols;
    }

    /**
     * 获取WebSocket的连接地址
     *
     * @param req HTTP请求对象
     * @return WebSocket的连接地址
     */
    private static String getWebSocketLocation(FullHttpRequest req) {
        String location = req.headers().get(HttpHeaderNames.HOST) + req.uri();
        return "wss://" + location;
    }

}
